package com.atguigu.realtime.app.dwm;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.app.BaseApp;
import com.atguigu.realtime.common.ConstantTopic;
import com.atguigu.realtime.util.MyKafkaUtil;
import com.google.common.collect.Lists;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.Date;

/**
 * @ClassName: DWMUVApp
 * @Description:
 * @Author: kele
 * @Date: 2021/4/20 10:49
 *
 * 初略统计UV，
 *
 **/
public class DWMUVApp extends BaseApp{

    public static void main(String[] args) {

        new DWMUVApp().init(30000,2,"DWMUVApp",ConstantTopic.DWD_PAGE,"DWMUVApp");

    }

    @Override
    protected void run(StreamExecutionEnvironment env, DataStreamSource<String> dataStreamSource) {

        //统计某个用户某天的第一条数据
        DataStream<JSONObject> processData = computerMid(dataStreamSource);

        processData.print();

        send2Kafka(processData);

    }

    //将数据发送到kafka中
    private void send2Kafka(DataStream<JSONObject> processData) {

        processData
                .map(JSON::toString)
                .addSink(MyKafkaUtil.getKafkaSink(ConstantTopic.DWM_UV));
    }

    /**
     * 每天，每个用户的第一条数据
     * @param stream
     *
     * 如果根据启动日志确定，只能确定从app端启动的用户
     * 根据页面日志确定用户
     */
    private DataStream<JSONObject> computerMid(DataStreamSource<String> stream) {

        SingleOutputStreamOperator<JSONObject> result = stream
                .map(JSON::parseObject)
                //添加水印，乱序时间是2s
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                                .withTimestampAssigner((element, recordTimestamp) -> element.getLong("ts"))
                )
                .keyBy(ele -> ele.getJSONObject("common").getString("mid"))
                //开窗，5s一个窗口
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                //
                .process(new ProcessWindowFunction<JSONObject, JSONObject, String, TimeWindow>() {

                    private SimpleDateFormat dateFormat;
                    private ValueState<Long> firstViewState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        firstViewState = getRuntimeContext()
                                .getState(new ValueStateDescriptor<Long>("firstViewState", Long.class));

                        dateFormat = new SimpleDateFormat("yyyy-MM-dd");
                    }

                    @Override
                    public void process(String key,
                                        Context context,
                                        Iterable<JSONObject> elements,
                                        Collector<JSONObject> out) throws Exception {
                        /**
                         * 将第一个窗口到的数据进行排序，取出第一条，之后的数据不需要
                         *
                         * 第二天的数据：如果此时到达的数据的日期（取窗口的开始时间作为数据日期）与状态中的日期不一致，重新统计
                         */
                        long start = context.window().getStart();
                        String nowDataTime = dateFormat.format(start);

                        //如果是第一条数据或者第二天的数据的第一个窗口
                        if (firstViewState.value() == null ||
                                !nowDataTime
                                        .equalsIgnoreCase(dateFormat.format(new Date(firstViewState.value())))) {

                            //google的方法，可将Iterable类型的数据转化为list集合
                            ArrayList<JSONObject> list = Lists.newArrayList(elements);

                        /*需要一个个比较，效率低
                        //list.sort((o1, o2) -> o1.getLong("ts").compareTo(o2.getLong("ts")));
                        list.sort(Comparator.comparing(o -> o.getLong("ts")));
                        JSONObject firstData = list.get(0); //取出最小值
                        */

                            //使用collecion工具类求最小值效率较高
                            JSONObject fistData =
                                    Collections
                                            .min(list, Comparator.comparing(o -> o.getLong("ts")));

                            firstViewState.update(fistData.getLong("ts"));

                            out.collect(fistData);
                        }


                    }
                });

        return result;

    }
}
